package test.top2.com.atguigu.utils;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Properties;

/**
 * ClassName: MyKafkaUtil
 * Package: test.top2.com.atguigu.utils
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/4/27 18:14
 * @Version 1.0
 */
public class MyKafkaUtil {

    private  static  final String BOOTSTRAP_SERVER = "hadoop102:9092";
    public static KafkaSource<String> getKafkaSource(String topicId,String groupId){
        return  KafkaSource.<String>builder()
                .setBootstrapServers(BOOTSTRAP_SERVER)
                .setTopics(topicId)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null){
                            return  new String(message);
                        }else {
                            return null;
                        }
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                })
                .setStartingOffsets(OffsetsInitializer.latest())
                .build();
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic){
        return new FlinkKafkaProducer<String>(BOOTSTRAP_SERVER,topic,new SimpleStringSchema());
    }

    public static FlinkKafkaProducer<JSONObject> getFlinkKafkaProducer(){

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BOOTSTRAP_SERVER);

        return  new FlinkKafkaProducer<JSONObject>("dwd_default_topic", new KafkaSerializationSchema<JSONObject>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                return  new ProducerRecord<>(jsonObject.getString("sink_table"),
                        jsonObject.getString("data").getBytes());

            }
        },properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

    }

    public static <T> FlinkKafkaProducer<T> getFlinkKafkaProducer(KafkaSerializationSchema<T> kafkaSerializationSchema) {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<T>(
                "dwd_default_topic",
                kafkaSerializationSchema,
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    public static String getTopicDbDDL(String groupId){
        return "create table topic_db(\n" +
                "   `database` string,\n" +
                "   `table` string,\n" +
                "   `type` string,\n" +
                "   `data` map<string,string>,\n" +
                "   `old` map<string,string>\n" +
                "   `pt` as proctime()\n" +
                ") " + getKafkaSourceConnOption("topic_db",groupId);
    }

    public static String getKafkaSourceConnOption(String topicId,String groupId){
        return "with(\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '"+topicId+"',\n" +
                "  'properties.bootstrap.servers' = '"+BOOTSTRAP_SERVER+"',\n" +
                "  'properties.group.id' = '"+groupId+"',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    public static String getKafkaSinkConnOption(String topic) {
        return "with ( " +
                "    'connector' = 'kafka', " +
                "    'topic' = '" + topic + "', " +
                "    'properties.bootstrap.servers' = '" + BOOTSTRAP_SERVER + "', " +
                "    'format' = 'json' " +
                ")";
    }

    public static String getKafkaUpsertSinkConnOption(String topicId){
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '"+topicId+"',\n" +
                "  'properties.bootstrap.servers' = '"+BOOTSTRAP_SERVER+"',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }
}
